package com.atguigu.Flink.timeAndwindow;

import com.atguigu.Flink.POJO.Event;
import com.atguigu.Flink.function.ClickSource;
import org.apache.flink.api.common.eventtime.*;
import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;

public class Flink01_UserDefineWaterMark {
    public static void main(String[] args) {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);
        //DataStreamSource<Event> ds = env.addSource(new ClickSource());
        SingleOutputStreamOperator<Event> ds = env.socketTextStream("hadoop102", 8888).map(new MapFunction<String, Event>() {
            @Override
            public Event map(String s) throws Exception {
                String[] split = s.split(",");
                return new Event(split[0], split[1], Long.valueOf(split[2]));
            }
        });
        //生成水位线
        ds.assignTimestampsAndWatermarks(new WatermarkStrategy<Event>() {

            //创建水位线生成对象
            @Override
            public WatermarkGenerator<Event> createWatermarkGenerator(WatermarkGeneratorSupplier.Context context) {
                return new WatermarkGenerator<Event>() {
                    private Long deply = 2000L;
                    private Long maxTs = Long.MIN_VALUE + deply;
                    //每条数据都会生成水位线
                    @Override
                    public void onEvent(Event event, long l, WatermarkOutput watermarkOutput) {
                        //有序流每条数据都生成水位线
//                        Watermark watermark = new Watermark(l);
//                        watermarkOutput.emitWatermark(watermark);
//                        System.out.println("为每条数据发射水位线"+watermark);
                        //maxTs = event.getTs();

                        //乱序流中每条数据生成水位线，要维护当前到达的数据最大的时间
//                        maxTs = Math.max(maxTs,l);
//                        Watermark watermark = new Watermark(maxTs);
//                        watermarkOutput.emitWatermark(watermark);
//                        System.out.println("有序流每条数据生成水位线"+watermark);

                        maxTs = Math.max(maxTs,l);
                    }
                    //周期执行，默认200ms
                    @Override
                    public void onPeriodicEmit(WatermarkOutput watermarkOutput) {
                        //有序流周期性发射水位线
//                        Watermark watermark = new Watermark(maxTs);
//                        watermarkOutput.emitWatermark(watermark);
//                        System.out.println("有序流周期性发射水位线"+watermark);

                        //乱序流周期性发射水位线
                        Watermark watermark = new Watermark(maxTs - deply);
                        watermarkOutput.emitWatermark(watermark);
                        System.out.println("有序流周期性发射水位线"+watermark);
                    }
                };
            }
            //创建时间戳生成对象
            @Override
            public TimestampAssigner<Event> createTimestampAssigner(TimestampAssignerSupplier.Context context) {
                return new TimestampAssigner<Event>() {
                    @Override
                    public long extractTimestamp(Event event, long l) {
                        return event.getTs();
                    }
                };
            }
        }).print();


        try {
            env.execute();
        } catch (Exception e) {
            throw new RuntimeException(e);
        }
    }
}
